/*
This file is part of [aconnect] library. 

Author: Artem Kustikov (kustikoff[at]tut.by)


This code is provided 'as-is', without any express or implied
warranty. In no event will the authors be held liable for any
damages arising from the use of this code.

Permission is granted to anyone to use this code for any
purpose, including commercial applications, and to alter it and
redistribute it freely, subject to the following restrictions:

1. The origin of this code must not be misrepresented; you must
not claim that you wrote the original code. If you use this
code in a product, an acknowledgment in the product documentation
would be appreciated but is not required.

2. Altered source versions must be plainly marked as such, and
must not be misrepresented as being the original code.

3. This notice may not be removed or altered from any source
distribution.
*/

#ifndef ACONNECT_THREAD_POOL_H
#define ACONNECT_THREAD_POOL_H

#include "types.hpp"

#include <boost/utility.hpp>
#include <boost/thread.hpp>
#include <boost/thread/condition.hpp>
#include <boost/detail/atomic_count.hpp>

#include <queue>
#include <stdexcept>

#include "logger.hpp"
#include "util.hpp"
#include "util.time.hpp"
#include "istoppable.hpp"

#include <boost/function.hpp>

namespace aconnect 
{
	template <typename P, typename Container>
	struct TopLoader
	{
		P top (Container &container)
		{
			P item = container.front();
			container.pop();

			return item;
		};
	};

	template <typename P>
	struct TopLoader <P, std::priority_queue<P> >
	{
		P top (std::priority_queue<P>& container)
		{
			P item = container.top();
			container.pop();
			return item;
		};
	};

	struct pool_started_error : public std::runtime_error 
	{
		pool_started_error() : std::runtime_error ("Pool is started...") { }
	};

	struct pool_queue_full_error : public std::runtime_error 
	{
		pool_queue_full_error() : std::runtime_error ("Pool queue overloaded") { }
	};
	

	template<class P, class Container> struct ThreadPoolWorkerAdapter;


	//==================================================================//
	//	class ThreadPool												//
	//==================================================================//
	template <typename P, 
		typename Container = std::queue<P> >
	class ThreadPool : public IStoppable
	{

	public:
		typedef boost::function1<void, const P&> worker_thread_proc;
		typedef boost::function1<void, const P&> check_item_proc;
		typedef ThreadPoolWorkerAdapter<P, Container> adapter_type;

		friend struct ThreadPoolWorkerAdapter<P, Container>;

	//==================================================================//
	private:
		Logger     *_logger;
		
		worker_thread_proc _workerProc;
		check_item_proc _checkProcessedItemProc;
		check_item_proc _checkQueuedItemProc;

		std::auto_ptr<boost::thread_group> _threadGroup;
		
		size_t	_maxWorkersCount;
		size_t	_maxQueuedRequestsCount;
		int	_poolRecycleTimeoutSec;			// sec
		int	_poolRecycleTimeoutNanoSec;		// nanosec - boost knows


		boost::detail::atomic_count _pendingWorkersCount;
		boost::detail::atomic_count _workersCount; // wrapper for _threadGroup.size()
        		
		boost::mutex _mainMutex;
		
		boost::mutex _queueMutex;
		boost::condition _queueCondition;
		
		Container _requestsQueue;
		bool _isWait;
		bool _isStopped;

		TopLoader<P, Container> _topLoader;

	public:
		ThreadPool (worker_thread_proc worker_proc = 0, int maxWorkersCount = 50, Logger *logger = 0) : 
			_logger (logger),
		    _workerProc (worker_proc),
			_checkProcessedItemProc (0),
			_checkQueuedItemProc (0),

			_maxWorkersCount (maxWorkersCount),
			_maxQueuedRequestsCount (maxWorkersCount * 2), // some heuristics
			_poolRecycleTimeoutSec (1),
			_poolRecycleTimeoutNanoSec (0),
			
			_pendingWorkersCount (0),
			_workersCount(0),
			
			_isWait (false),
			_isStopped (false)
		{
			_threadGroup.reset(new boost::thread_group());

		}
		
		//==================================================================//
		//	Public interface												//
		//==================================================================//

		//==================================================================//
		//	IStoppable implementation										//
		//==================================================================//
		bool isStopped()	
		{ 
			boost::mutex::scoped_lock lock (_mainMutex);
			return _isStopped;
		}
		
		//==================================================================//
		// ThreadPool implementation										//
		//==================================================================//
		void process (P item);

		// prioritized processing
		void addItemToProcess (P item) throw (pool_started_error);

		void startProcessing (size_t delay = (size_t) -1);

		void stop (bool joinAll);
		
		void wait ();
				
		
		void setLog (Logger *log)	{   
			_logger = log;  
		}
        
		Logger* log () const	{   
			return _logger;	
		}   
				
		void setMaxWorkersCount(int count) {	
			boost::mutex::scoped_lock lock (_mainMutex);

			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_maxWorkersCount = count;		
		} 

		int maxWorkersCount() const	{	
			return _maxWorkersCount;		
		} 

		void setMaxQueuedRequestsCount(size_t count) {	
			boost::mutex::scoped_lock lock (_mainMutex);
			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_maxQueuedRequestsCount = count;		
		} 
	

		void setWorkerProc(worker_thread_proc p) {
			boost::mutex::scoped_lock lock (_mainMutex);

			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_workerProc = p;
		} 
		
		worker_thread_proc workerProc() const {	
			return _workerProc;				
		} 

		void setPoolRecycleTimeouts(int timeoutSec, int timeoutNanoSec = 0) 
		{
			boost::mutex::scoped_lock lock (_mainMutex);

			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_poolRecycleTimeoutSec = timeoutSec;
			_poolRecycleTimeoutNanoSec = timeoutNanoSec;
		} 

		
		//==================================================================//
		//		Callbacks													//
		//==================================================================//
		void setProcessedItemProc(check_item_proc p) {
			boost::mutex::scoped_lock lock (_mainMutex);

			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_checkProcessedItemProc = p;
		} 
		
		check_item_proc processedItemProc() const {	
			return _checkProcessedItemProc;				
		}

		void setQueuedItemProc (check_item_proc p) {
			boost::mutex::scoped_lock lock (_mainMutex);

			assert ((long) _workersCount == 0);
			assert ((long) _requestsQueue.empty());

			_checkQueuedItemProc = p;
		} 
		
		check_item_proc queuedItemProc() const {	
			return _checkQueuedItemProc;				
		}

		//==================================================================//
		
		long workersCount () {	
			return _workersCount; 
		}

		long pendingWorkersCount () {
			return _pendingWorkersCount; 
		}

		bool isRequestsQueueEmpty() {	
			return _requestsQueue.empty();	
		}

		// no thread-safe
		size_t requestsQueueLength() {
			return _requestsQueue.size();
		}

		//==================================================================//
		
	protected:

		void runWorkerThread (const P& item)
		{
			adapter_type adapter (*this, item);
			_threadGroup->create_thread (adapter);
		
			++_workersCount;

			boost::thread::yield ();
		}
	
		bool setStopped () {	
			boost::mutex::scoped_lock lock (_mainMutex);
			bool was = _isStopped;
			_isStopped = true;

			return was;
		}

		void releaseWorker () {	
            boost::mutex::scoped_lock lock (_mainMutex);
			assert (_isStopped == true || _isWait == true);

			long wc = --_workersCount;
			assert (wc >= 0 && "Negative workers count!");
		}
		
		bool waitRequest (P &item) 
		{
			bool result = true;

			try 
			{
				if (isStopped())
					return false;

				++_pendingWorkersCount;
							
				// wait for new request
				boost::mutex::scoped_lock lock (_queueMutex);
				
				while (!_queueCondition.timed_wait (lock, aconnect::util::createTimePeriod (_poolRecycleTimeoutSec, _poolRecycleTimeoutNanoSec)) )
				{
					if ( isStopped()) {
						result = false;
						break;
					}

					if (_isWait) {
						
						if (_requestsQueue.empty())
							result = false;
						
						break;
					}


					if (_pendingWorkersCount >= _workersCount && !_requestsQueue.empty())
						break;
				}

				if (isStopped())
					result = false;

				--_pendingWorkersCount;
				
				if (result) 
                {
					if (_requestsQueue.empty())
						return true; // wait again (((
					
					if (_checkProcessedItemProc)
						_checkProcessedItemProc (item);
					
					item = _topLoader.top(_requestsQueue); // get copy

					if (_checkQueuedItemProc)
						_checkQueuedItemProc (item);
				}
			
			
			} catch (std::exception &err) {
				if (_logger) _logger->error(err);
			
			} catch (...) {
				if (_logger) _logger->error("ThreadPool: processing item wait failed - unknown exception caught");
			}

			return result;
		}

	};

	//==================================================================//
	//	class ThreadPoolWithPriority									//
	//==================================================================//

	template<typename P, typename Comparer = std::less<P> >
	class ThreadPoolWithPriority : public ThreadPool<P,
			std::priority_queue<P, std::vector<P>, Comparer> > 
	{
	public:
		typedef boost::function1<void, const P&> worker_thread_proc;
		typedef boost::function1<void, const P&> check_item_proc;
		typedef std::priority_queue<P, std::vector<P>, Comparer> container_type;
		
	public:
	
		ThreadPoolWithPriority (worker_thread_proc worker_proc = 0, 
				int maxWorkersCount = 50, 
				Logger *logger = 0) 
					: ThreadPool<P, container_type> (worker_proc, maxWorkersCount, logger)
		{
			
		}
	};

	//==================================================================//
	//	class ThreadPoolWorkerAdapter									//
	//==================================================================//
	template <typename P, typename Container>
	struct ThreadPoolWorkerAdapter
	{
		typedef ThreadPool<P, Container> pool_type;

		ThreadPoolWorkerAdapter (pool_type& pool, P item /* get copy at adapter creation */) :
			_pool (pool),
			_item (item) { 
		}
		
		inline void operator () ()
		{
			do 
			{
				if (_pool.isStopped())
					break;
				
				try {
					// process request
					_pool.workerProc()(_item);
				
				} catch (socket_error &err) {
					if (_pool.log())
						_pool.log()->warn(err);
				
				} catch (std::exception &err) {
					if (_pool.log())
						_pool.log()->error(err);
				
				} catch (...) {
					if (_pool.log())
						_pool.log()->error("ThreadPool: thread proc execution failed - unknown exception caught");
				}

			
			} while (_pool.waitRequest (_item));

			_pool.releaseWorker();
		}

	private:
		pool_type& _pool;
		P _item;
	};

	//==================================================================//
	//	ThreadPool definitions											//
	//==================================================================//

	template<typename P, typename Container>
	void ThreadPool<P, Container>::process (P item) 
	{	
		assert (_threadGroup.get());

		{
			boost::mutex::scoped_lock lock (_mainMutex);
			
			if (_isStopped)
				return;
			
			if ((long) _workersCount < (long) _maxWorkersCount) {
				// create new worker thread
				runWorkerThread (item);
				return;
			}
        }
        
		boost::mutex::scoped_lock queueLock (_queueMutex);

		if (_requestsQueue.size() > _maxQueuedRequestsCount)
			throw pool_queue_full_error();
		
		// add request to queue
		_requestsQueue.push (item);
		_queueCondition.notify_one();
		
	}

	// prioritized processing
	template<typename P, typename Container>
	void ThreadPool<P, Container>::addItemToProcess (P item) throw (pool_started_error)
	{
		{
			boost::mutex::scoped_lock lock (_mainMutex);
			assert ((long) _workersCount == 0);
			
			if (_workersCount > 0)
				throw pool_started_error();
		}

		boost::mutex::scoped_lock lock (_queueMutex);
		_requestsQueue.push (item);
	}

	template<typename P, typename Container>
	void ThreadPool<P, Container>::startProcessing (size_t delay)
	{
		boost::mutex::scoped_lock lock (_mainMutex);
		assert ((long) _workersCount == 0);
		
		if (_workersCount > 0)
			throw pool_started_error();

		size_t count = _maxWorkersCount;
			
		while (count-- > 0) {
			if (_isStopped)
				return;
		
			boost::mutex::scoped_lock queueLock (_queueMutex);
			
			if (_requestsQueue.empty())
				break;

			P item = _topLoader.top(_requestsQueue); // get copy
			
			queueLock.unlock();
			
			// create new worker thread
			runWorkerThread (item);

			if (delay != (size_t) -1)
				aconnect::util::sleep (delay);
		}
	}

	template<typename P, typename Container>
	void ThreadPool<P, Container>::stop (bool joinAll)
	{
		bool wasStopped = setStopped ();
		
		if (_logger)
			_logger->debug ("Threads pool unloading, pending workers count: %d", (long) _pendingWorkersCount );

		while (_pendingWorkersCount > 0) 
		{
			boost::mutex::scoped_lock lock (_queueMutex);
			_queueCondition.notify_all();
		}

		if (_logger)
			_logger->debug ("Threads pool unloading, workers count: %d", (long) _workersCount );

		if (joinAll)
			_threadGroup->join_all();
		else if (!wasStopped)
			_threadGroup->interrupt_all();

		// clean all
		{
			boost::mutex::scoped_lock lock (_queueMutex);

			Container empty;
			std::swap( _requestsQueue, empty );
		}

		boost::mutex::scoped_lock lock (_mainMutex);

		_threadGroup.reset(new boost::thread_group());
		
		while (_workersCount > 0)
			--_workersCount;

		_isStopped = false;

	}

	template<typename P, typename Container>
	void ThreadPool<P, Container>::wait ()
	{
		{
			boost::mutex::scoped_lock lock (_queueMutex);
			_isWait = true;
		}

		_threadGroup->join_all();

		boost::mutex::scoped_lock lock (_mainMutex);
		// clean all
		_threadGroup.reset(new boost::thread_group());
		
		while (_workersCount > 0)
			--_workersCount;

		_isWait = false;
	}
			
	
}
#endif // ACONNECT_THREAD_POOL_H
